package com.hubiwei.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.hubiwei.gmall.realtime.annotation.NotSink;
import com.hubiwei.gmall.realtime.bean.TableProcess;
import com.hubiwei.gmall.realtime.bean.VisitorStats;
import com.hubiwei.gmall.realtime.common.Constant;
import com.hubiwei.gmall.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;


public class FlinkSinkUtil {
    public static SinkFunction<String> getKafkaSink(String topic) {
        
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                @Nullable Long timestamp) {
                    return new ProducerRecord<byte[], byte[]>(topic, element.getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                @Nullable Long timestamp) {
                    String topic = element.f1.getSink_table();
                    String data = element.f0.toJSONString();
                    return new ProducerRecord<>(topic, data.getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
        
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        
        return new PhoenixSink();
    }
    
    public static void main(String[] args) {
        getClickHouseSink("abc", "t", VisitorStats.class);
    }
    
    public static <T> SinkFunction<T> getClickHouseSink(String db,
                                                        String table,
                                                        Class<T> tClass) {
        String driver = Constant.CLICK_HOUSE_DRIVER;
        String url = Constant.CLICK_HOUSE_URL_PRE + db;
        
        StringBuilder sql = new StringBuilder();
        // 拼接sql语句:   T类型的属性名一定要和表的列名一致
        // insert into t(a,b,c) values(?,?,?)
        sql
            .append("insert into ")
            .append(table)
            .append("(");
        // to 拼接字段名
        Field[] fields = tClass.getDeclaredFields();
        for (Field field : fields) {
            // 如果这个属性有注解就不执行
            NotSink notSink = field.getAnnotation(NotSink.class);
            if (notSink == null) { // 如果为空,表示不存在这个注解, 需要把这个字段拼接到sql中
                sql.append(field.getName()).append(",");
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        
        sql.append(") values(");
        // 拼接 ?
        for (Field field : fields) {
            NotSink notSink = field.getAnnotation(NotSink.class);
            if (notSink == null) { // 如果为空,表示不存在这个注解, 需要把这个字段拼接到sql中
                sql.append("?,");
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(")");
        
        //        System.out.println(sql.toString());
        
        return getJdbcSink(driver, url, sql.toString());
    }
    
    public static <T> SinkFunction<T> getJdbcSink(String driver,
                                                  String url,
                                                  String sql) {
        
        return JdbcSink.sink(sql,
                             new JdbcStatementBuilder<T>() {
                                 @Override
                                 public void accept(PreparedStatement ps,
                                                    T t) throws SQLException {
                                     // 给占位符赋值 todo
                                     // 获取class:  1. 类名.class 2. 对象.getClass 3. Class.forName("类名")
                                     Field[] fields = t.getClass().getDeclaredFields();
                                     try {
                                         for (int i = 0, position = 1; i < fields.length; i++) {
                                             Field field = fields[i];
                                             NotSink notSink = field.getAnnotation(NotSink.class);
                                             if (notSink == null) {
                                                 field.setAccessible(true); //允许访问私有属性
                                                 ps.setObject(position++, field.get(t));  // t.属性  读取这个属性的值
                            
                                             }
                                         }
                                     } catch (IllegalAccessException e) {
                                         e.printStackTrace();
                                     }
                                 }
                             },
                             new JdbcExecutionOptions.Builder()
                                 .withBatchIntervalMs(1000)
                                 .withBatchSize(1024)
                                 .withMaxRetries(3)
                                 .build(),
                             new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                 .withDriverName(driver)
                                 .withUrl(url)
                                 .build()
        );
        
    }
    
}
